﻿#include"./Session.hpp"
#include"./Malloc/TypesDef.hpp"
#include<iostream>


namespace MMFNetwork {

    Session::Session(
            io_context& service,
            uint32_t bufferSize,
            const SessionErrorHandler& errorHandler,
            const SessionStatusHandler& statusHandler,
            const SessionPackageHandler& pkgHandler,
            MMFInetFamily family
        )
        : mService(service)
        , mSocket(service, family==MMFInetFamily::V4? ip::tcp::v4() : ip::tcp::v6())    //根据网络类型来创建
        , mBufferSize(bufferSize)
        , mCurrentRecvBytes(0)
        , mCurrentBuff(0)           ///< 当前使用的缓冲
        , mRecvBuffers(4)           ///< 接收缓冲数
        , mErrorHandler(errorHandler)
        , mStatusHandler(statusHandler)
        , mPkgHandler(pkgHandler)
        , mStatus(MMFNetwork::Status::Invalid)
    {
        for (uint32_t i = 0; i < mRecvBuffers.size(); i++)
            mRecvBuffers[i] = aligned_alloc(PAGE_SIZE, bufferSize);
    }

    Session::~Session()
    {
        for (uint32_t i = 0; i < mRecvBuffers.size(); i++)
            aligned_free(mRecvBuffers[i]);
    }

    void Session::Start()
    {
        boost::system::error_code ec;
        mSocket.set_option(socket_base::linger(true, 30), ec);      //同步套接字下，关闭套接字的时候我方需要等到所有数据发送完毕，异步会返回-1，需要getlasterror判断是否出错
        mSocket.set_option(socket_base::keep_alive(true), ec);
        mSocket.set_option(ip::tcp::no_delay(true), ec);            //不启用nigle算法，延迟更低减少粘包但是会降低传输效率

        mLocalIp = mSocket.local_endpoint(ec).address().to_string();
        mRemoteIp = mSocket.remote_endpoint(ec).address().to_string();
        mLocalPort = mSocket.local_endpoint(ec).port();
        mRemotePort = mSocket.remote_endpoint(ec).port();

        if (ec)
        {
            mErrorHandler(shared_from_this(), ec);
            return;
        }

        //开始异步收包
        AsyncRecvPackage();
    }

    const std::string& Session::LocalIp()
    {
        return mLocalIp;
    }

    const std::string& Session::RemoteIp()
    {
        return mRemoteIp;
    }

    uint16_t Session::LocalPort()
    {
        return mLocalPort;
    }

    uint16_t Session::RemotePort()
    {
        return mRemotePort;
    }

    Status Session::CurrentStatus()
    {
        return mStatus;
    }

    void Session::Stop(MMFNetwork::Status status)
    {
        if (!mSocket.is_open())
            return;

        try
        {
            mSocket.cancel();
            mSocket.shutdown(socket_base::shutdown_type::shutdown_both);
            mSocket.close();
        }
        catch (...)
        {
            return;
        }

        mStatus = status;
        mStatusHandler(shared_from_this(), mStatus);
    }

    bool Session::Send(void* buff, uint32_t count)
    {
        boost::system::error_code ec;
        boost::asio::write(mSocket, boost::asio::buffer(buff, count), ec);
        if (ec)
        {   
            Stop(MMFNetwork::Status::WriteErr);
            return false;
        }

        return true;
    }

    bool Session::Recv(void* buff, uint32_t count)
    {
        boost::system::error_code ec;
        boost::asio::read(mSocket, boost::asio::buffer(buff, count), ec);
        if (ec)
        {   
            Stop(MMFNetwork::Status::ReadErr);
            return false;
        }

        return true;
    }

    void Session::AsyncSend(void* buff, uint32_t count)
    {
        boost::asio::async_write(mSocket, boost::asio::buffer(buff, count), 
        [this, count, buff](boost::system::error_code ec, size_t sentNum){
            if (ec || sentNum < count)  //发送出错或者发送字节数与数据字节数不一致
                Stop(MMFNetwork::Status::AsyncWriteErr);
        });
    }

    void Session::AsyncRecv(void* buff, uint32_t count, const SessionAsyncHandler& handler)
    {
        boost::asio::async_read(mSocket, boost::asio::buffer(buff, count),
        [this, count, buff, handler](boost::system::error_code ec, size_t sentNum){
            if (ec || sentNum < count)  //接收出错或者接收字节数与数据字节数不一致
                Stop(MMFNetwork::Status::AsyncReadErr);
            else
                handler(buff, true);
        });
    }

    void Session::AsyncRecvPackage()
    {
        //假定需要收取这么大
        size_t recvBytes = mBufferSize - mCurrentRecvBytes;
        void* recvBuff = mRecvBuffers[mCurrentBuff];

        mSocket.async_receive(
            boost::asio::buffer((char*)recvBuff + mCurrentRecvBytes, recvBytes), //接着收数据
            [this, recvBuff](boost::system::error_code ec, size_t num){

                //取消操作的时候会立即返回并且错误码为boost::asio::error::operation_aborted
                //这时候不需要做任何操作
                if (ec == boost::asio::error::operation_aborted)
                    return;
                else if (ec)  //接收出错
                {
                    Stop(MMFNetwork::Status::AsyncReadErr);
                    return;
                }

                //增加接收数量
                mCurrentRecvBytes += (uint32_t)num;

                //判断是否收满了头
                if (mCurrentRecvBytes > sizeof(MMFMessageHeader))
                {
                    //转换指针类型
                    MMFMessageHeader* head = (MMFMessageHeader*)recvBuff;

                    //判断数据是否收满
                    uint32_t dataLen = head->mDataBytes;

                    //如果数据已经收满
                    if (mCurrentRecvBytes >= dataLen)
                    {
                        //处理多个消息被收满的情况
                        uint32_t next = 0;
                        do 
                        {
                            //已经收的数据是一个完整的包
                            mCurrentRecvBytes = mCurrentRecvBytes - dataLen;

                            //通知包数据已经收完
                            mPkgHandler(shared_from_this(), (char*)recvBuff + next, dataLen);
                            next += dataLen;

                            //如果不够头长度说明，没有消息
                            if (mCurrentRecvBytes < sizeof(MMFMessageHeader))
                                break;
                            
                            //计算下一个数据的位置
                            head = (MMFMessageHeader*)((char*)recvBuff + next);
                            dataLen = head->mDataBytes;

                        }while (mCurrentRecvBytes >= dataLen);

                        //交换接收缓冲缓冲
                        mCurrentBuff++;
                        mCurrentBuff %= mRecvBuffers.size();
                        void* newBuff = mRecvBuffers[mCurrentBuff];

                        //拷贝下一个数据
                        if (mCurrentRecvBytes > 0)
                            memcpy(newBuff, (char*)recvBuff + dataLen, mCurrentRecvBytes);
                    }
                }

                AsyncRecvPackage();
            }
        );
    }
}